package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskRefreshService;
import com.heima.schedule.service.TaskService;
import io.jsonwebtoken.lang.Collections;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {


    @Override
    public long addTask(Task task) {

        //1.添加任务数据到DB中
        boolean flag = addTaskToDB(task);

        //2.添加任务数据到Redis中
        if(flag){
            addTaskToCache(task);
        }
        return task.getTaskId();
    }

    @Autowired
    private CacheService cacheService;

    //添加任务数据到Redis中
    private void addTaskToCache(Task task) {

        //1.获取当前系统时间、预设时间、任务执行时间
        long currentTime = DateTime.now().getMillis(); //系统当前时间
        long futureTime = DateTime.now().plusMinutes(5).getMillis();//预设时间
        long executeTime = task.getExecuteTime();//任务执行时间

        //2.如果任务执行时间<=系统当前时间，则将任务添加到Redis的当前任务队列中（list）
        if(executeTime <= currentTime){
            String redisKeyPostFix = task.getTaskType() + ":" + task.getPriority();
            String redisKey = ScheduleConstants.TOPIC + redisKeyPostFix; //最终rediskey的组成结构：  topic:1:100
            cacheService.lLeftPush(redisKey, JSON.toJSONString(task));

            log.info("任务添加到当前任务队列成功，任务ID：{}", task.getTaskId());
        } else {
            //3. 系统当前时间<如果任务执行时间<预设时间，则将任务添加到Redis的未来任务队列中（zset）
            if(executeTime<=futureTime){
                String redisKeyPostFix = task.getTaskType() + ":" + task.getPriority();
                String redisKey = ScheduleConstants.FUTURE + redisKeyPostFix; //最终rediskey的组成结构：  future:1:100
                cacheService.zAdd(redisKey,JSON.toJSONString(task), executeTime);
                log.info("任务添加到未来任务队列成功，任务ID：{}", task.getTaskId());
            } else {
                log.error("忽略当前任务，不会添加到未来任务队列，任务ID：{}", task.getTaskId());
            }
        }
    }

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    //添加任务数据到DB中
    private boolean addTaskToDB(Task task) {
        try {
            //1.复制task给taskInfo
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));

            //2.保存task_info数据
            taskinfoMapper.insert(taskinfo);
            task.setTaskId(taskinfo.getTaskId());

            //3.复制taskInfo给taskInfoLogs
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo, taskinfoLogs);
            taskinfoLogs.setVersion(1);
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);

            //4.保存task_info_logs数据
            taskinfoLogsMapper.insert(taskinfoLogs);

            return true;
        } catch (BeansException e) {
            e.printStackTrace();
        }

        return false;
    }


    @Autowired
    private TaskRefreshService taskRefreshService;

    /**
     * 同步未来任务队列数据到当前任务队列中
     */
    @Scheduled(cron = "*/10 * * * * ?")
    public void refreshCache(){

        String setNxLockName = "REFRESH_CACHE";

        String lock = cacheService.tryLock(setNxLockName, 10 * 1000);

        if(StringUtils.isNotEmpty(lock)){
            log.info("[refreshCache]获取到分布式锁，开始执行定时刷新Redis队列");

            //1.获取Redis中全部未来队列的key
            Set<String> keySet = cacheService.scan(ScheduleConstants.FUTURE + "*");
            //2.遍历key的集合
            if(!keySet.isEmpty()){
                for (String zsetKey : keySet) { //zsetkey的结构： future:1:100

                    //提交异步刷新Redis缓存队列的任务
                    taskRefreshService.refreshCache(zsetKey);
                }
            }
        } else {
            log.info("[refreshCache]未获取到分布式锁，开始执行定时刷新Redis队列");
        }



    }


    /**
     * 系统重新启动后，立即同步表中的待执行任务到Redis中
     */
    @PostConstruct
    public void reloadDataToRedis(){
        String lockName = "RELOADDATATOREDIS";
        String lock = cacheService.tryLock(lockName, 10 * 1000);

        if(StringUtils.isNotEmpty(lock)){
            log.info("[reloadDataToRedis]获取到分布式锁，立即开始同步表中数据到Redis");

            //1.清空redis中未来任务队列和当前任务队列数据

            //获取到全部当前任务队列的redis的key
            Set<String> topicRedisKeySet = cacheService.scan(ScheduleConstants.TOPIC + "*");
            if(!topicRedisKeySet.isEmpty()){
                cacheService.delete(topicRedisKeySet);
            }

            //获取到全部未来任务队列的redis的key
            Set<String> futureRedisKeySet = cacheService.scan(ScheduleConstants.FUTURE + "*");
            if(!futureRedisKeySet.isEmpty()){
                cacheService.delete(futureRedisKeySet);
            }

            //2.从任务日志表查询所有待执行的任务
            List<TaskinfoLogs> taskinfoLogsList = taskinfoLogsMapper.selectList(Wrappers.<TaskinfoLogs>lambdaQuery().eq(TaskinfoLogs::getStatus, ScheduleConstants.SCHEDULED));
            if(!Collections.isEmpty(taskinfoLogsList)){
                for (TaskinfoLogs taskinfoLogs : taskinfoLogsList) {
                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfoLogs, task);
                    task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());

                    //3.将任务添加到Redis的未来任务队列和当前任务队列中
                    this.addTaskToCache(task);
                }
            }
        } else {
            log.error("[reloadDataToRedis]未获取到分布式锁");
        }
    }


    /**
     * 按照任务和优先级拉取当前任务队列中任务
     * @param type
     * @param priority
     * @return
     */
    @Override
    public Task poll(int type, int priority) {
        //1.根据type和priority组装redisKey
        String redisKey = ScheduleConstants.TOPIC + type + ":" + priority;

        //2.根据redisKey从当前任务队列（LIST）中拉取任务（右侧POP任务）
        String taskJSON = cacheService.lRightPop(redisKey);

        Task task = null;
        if(StringUtils.isNotEmpty(taskJSON)){
            task = JSON.parseObject(taskJSON, Task.class);

            byte[] parameters = task.getParameters();


            //3.删除task_info表的记录
            taskinfoMapper.deleteById(task.getTaskId());

            //4.更新task_info_logs表记录状态为已执行（乐观锁会自动将version当做条件并更新version+1）
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            /**
             * 最终执行的SQL，类似于：
             * ==>  Preparing: UPDATE taskinfo_logs SET version=?, status=? WHERE task_id=? AND version=?
             * ==> Parameters: 2(Integer), 1(Integer), 1478026992393125890(Long), 1(Integer)
             * <==    Updates: 1
             */
            taskinfoLogs.setTaskId(task.getTaskId()); //更新条件1
            taskinfoLogs.setVersion(1); //更新条件2
            taskinfoLogs.setStatus(ScheduleConstants.EXECUTED); //要更新的值
            taskinfoLogsMapper.updateById(taskinfoLogs);
        }
        return task;
    }
}
